package seven.two;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import seven.entity.FlowBean;

public class FlowCountTwo {
	
	static class FlowCountMapper extends Mapper<LongWritable,Text,Text,FlowBean>{

		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, FlowBean>.Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			String[] files = line.split("\t");
			String phoneNbr = files[0];
			long upFlow = Long.parseLong(files[1]);
			long downFlow = Long.parseLong(files[2]);
			context.write(new Text(phoneNbr), new FlowBean(upFlow,downFlow));
		} 
	}
	
	static class FlowCountReducer extends Reducer<Text,FlowBean,Text,FlowBean>{

		@Override
		protected void reduce(Text key, Iterable<FlowBean> values, Reducer<Text, FlowBean, Text, FlowBean>.Context context)
				throws IOException, InterruptedException {
			 
			 long sumUpFlow = 0;
			 long sumDownFlow = 0;
			 for(FlowBean bean : values){
				 sumUpFlow += bean.getUpFlow();
				 sumDownFlow += bean.getDownFlow();
			 }
			 FlowBean resultBean = new FlowBean(sumUpFlow,sumDownFlow);
			 context.write(key, resultBean);
		} 
	}
	//args是cmd运行时传入的参数
	public static void main(String[] args)throws Exception {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);
		//指定本程序的jar包所在的本地路径
		job.setJarByClass(FlowCountTwo.class);
		//指定mapper/reducer业务类
		job.setMapperClass(FlowCountMapper.class);
		job.setReducerClass(FlowCountReducer.class);
		
		//指定自己的数据分区器
		job.setPartitionerClass(ProvincePartitioner.class);
		
		//指定mapper输出的kv类型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(FlowBean.class);
		//指定最终输入的kv类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(FlowBean.class);
		//指定job的输入原始文件所在目录
		//FileInputFormat.setInputPaths(job,new Path(args[0]));
		FileInputFormat.setInputPaths(job,new Path("E:/code/workspace/hadoopone/src/main/java/seven/entity/input"));
		//指定job输出结果目录所在
		FileOutputFormat.setOutputPath(job, new Path("E:/code/workspace/hadoopone/src/main/java/seven/entity/output"));
		//将job中配置的相关参数，以及job所用的java类所在jar包提交给yarn去运行。
		boolean res = job.waitForCompletion(true);
		System.out.println(res?0:1);
		
	}
	
}
